package com.example.flinkcourse.lesson1.source;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.hbase.source.HBaseSource;
import org.apache.flink.connector.hbase.source.HBaseSourceBuilder;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * HBase 数据源工厂类
 * 用于创建 HBase 数据源
 */
public class HBaseSourceFactory {
    private static final Logger LOG = LoggerFactory.getLogger(HBaseSourceFactory.class);

    /**
     * 创建 HBase 数据源
     * @return HBaseSource<String>
     */
    public static HBaseSource<String> createHBaseSource() {
        LOG.info("Creating HBase source...");

        // 创建 HBase 配置
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "localhost");
        conf.set("hbase.zookeeper.property.clientPort", "2181");

        // 如果启用了 Kerberos 认证
        if (isKerberosEnabled()) {
            configureKerberos(conf);
        }

        // 创建 Scan 对象
        Scan scan = new Scan();
        scan.setCaching(1000);
        scan.setCacheBlocks(false);

        // 创建 HBase 数据源
        return HBaseSourceBuilder.<String>builder()
            .setTableName(TableName.valueOf("input_table"))
            .setScan(scan)
            .setDeserializer(new SimpleStringSchema())
            .setConfiguration(conf)
            .build();
    }

    /**
     * 检查是否启用 Kerberos 认证
     */
    private static boolean isKerberosEnabled() {
        // 从配置文件中读取是否启用 Kerberos
        return false; // 默认不启用
    }

    /**
     * 配置 Kerberos 认证
     */
    private static void configureKerberos(Configuration conf) {
        try {
            conf.set("hbase.security.authentication", "kerberos");
            conf.set("hbase.master.kerberos.principal", "hbase/_HOST@EXAMPLE.COM");
            conf.set("hbase.regionserver.kerberos.principal", "hbase/_HOST@EXAMPLE.COM");
            
            UserGroupInformation.setConfiguration(conf);
            UserGroupInformation.loginUserFromKeytab("hbase/localhost@EXAMPLE.COM", "/path/to/hbase.keytab");
        } catch (Exception e) {
            LOG.error("Failed to configure Kerberos authentication", e);
            throw new RuntimeException("Failed to configure Kerberos authentication", e);
        }
    }
} 